#include "rpcprovider.h"
#include "mprpcapplication.h"
#include "rpcheader.pb.h"
#include "logger.h"
#include "zookeeperutil.h"

/**
 * service_name => service描述
 *                      对应一个 service*，记录对象（一个对象可能有很多个方法），服务方法都是对象的方法，最后都是对象来调用的
 *                      method_name => method 方法对象
*/
/**
 * 这里是框架提供给外部使用的，可以发布 rpc 方法的函数接口
 * 这里的参数用基类 Service 的指针接收，框架不能依赖业务
 * 
 * 下面是 main 函数中使用的例子
 * // provider 是一个 rpc 网络服务对象，把 UserService 对象发布到 rpc 节点上
 * RpcProvider provider;
 * provider.NotifyService(new UserService);
*/
void RpcProvider::NotifyService(google::protobuf::Service *service)
{
    ServiceInfo service_info;

    // 获取服务对象的描述信息
    const google::protobuf::ServiceDescriptor *pserviceDesc = service->GetDescriptor();
    // 获取服务的名字
    std::string service_name = pserviceDesc->name();  // .proto 中定义的 service 名字
    // 获取服务方面对象 service 的方法的数量
    int methodCnt = pserviceDesc->method_count();

    // std::cout << "service_name: " << service_name << std::endl;
    LOG_INFO("service_name: %s", service_name.c_str());

    for(int i=0; i<methodCnt; i++)
    {
        // 获取服务对象执行下标的服务方法的描述（抽象描述）
        const google::protobuf::MethodDescriptor* pmethodDesc = pserviceDesc->method(i);
        std::string method_name = pmethodDesc->name();  // .proto 中 service 包含的 method 名字
        service_info.m_methodMap.insert({method_name, pmethodDesc});

        // std::cout << "method_name: " << method_name << std::endl;
        LOG_INFO("method_name: %s", method_name.c_str());
    }

    service_info.m_service = service;
    m_serviceMap.insert({service_name, service_info});
}

// 启动 rpc 服务节点，开始提供 rpc 远程调用（muduo 服务器也在这里定义、启动）
void RpcProvider::Run()
{
    // 读取配置文件中 rpc server的信息
    std::string ip = MprpcApplication::GetInstance().GetConfig().Load("rpcserverip");  // rpc server ip
    uint16_t port = atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcserverport").c_str());  // rpc server port; atoi 要的是 char*
    muduo::net::InetAddress address(ip, port);

    // 创建 TcpServer 对象
    muduo::net::TcpServer server(&m_eventLoop, address, "RpcProvider");
    // 绑定连接回调和消息读写回调方法
    server.setConnectionCallback(std::bind(&RpcProvider::OnConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcProvider::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

    // 设置 muduo 库的线程数量
    server.setThreadNum(4);

    // 把当前 rpc 节点上要发布的服务全部注册到 zk 上，让 rpc client 可以从 zk 上发现服务
    // session timeout 30s    zkclient 网络IO线程 1/3的timeout时间发送ping心跳消息
    ZkClient zkCli;
    zkCli.Start();  // 连接zk server
    // service_name 为永久性节点  method_name 为临时性节点
    for(auto &sp: m_serviceMap)
    {
        // /service_name    /UserServiceRpc
        std::string service_path = "/" + sp.first;
        zkCli.Create(service_path.c_str(), nullptr, 0);

        for(auto &mp: sp.second.m_methodMap)
        {
            // /service_name/method_name    /UserServiceRpc/Login 存储当前这个 rpc 服务节点主机的 ip 和 port
            std::string method_path = service_path + "/" + mp.first;
            char method_path_data[128] = {0};
            sprintf(method_path_data, "%s:%d", ip.c_str(), port);
            // ZOO_EPHEMERAL 表示临时性节点
            zkCli.Create(method_path.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);
        }
    }

    // rpc 服务端准备启动，打印信息
    std::cout << "RpcProvider start service at ip:" << ip << " port:" << port << std::endl;

    // 启动网络服务
    server.start();
    m_eventLoop.loop();
}

// 新的socket连接回调
void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr& conn)
{
    if(!conn->connected())
    {
        // 和 rpc client 的连接断开了
        conn->shutdown();  // 关闭文件描述符
    }
}

/**
 * 在框架内部，RpcProvider 和 RpcConsumer 协商好之间通信用的 protobuf 数据类型
 * service_name method_name args  定义 proto 的 message 类型，进行数据头（除了参数具体数据）的序列化和反序列化
 *                                  service_name method_name args_len  防止TCP粘包问题
 * header_size（四个字节）  header_str  args_size args_str
*/
// 已建立连接用户的读写事件回调 如果远程有 rpc 服务的调用请求，onMessage 方法就会响应
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr& conn, muduo::net::Buffer* buffer, muduo::Timestamp time)
{
    // 网络上接收的远程 rpc 调用的请求字符流  方法名、参数
    std::string recv_buf = buffer->retrieveAllAsString();

    // 从字符流中读取前四个字节的内容
    uint32_t header_size = 0;
    recv_buf.copy((char*)&header_size, 4, 0);  // 从第0个字符开始copy4个字节

    // 根据 header_size 读取数据头的原始字符流(service_name method_name args_len) → 反序列化数据得到rpc请求的详细信息
    std::string rpc_header_str = recv_buf.substr(4, header_size);
    mprpc::RpcHeader rpcHeader;
    std::string service_name;  // 服务对象名
    std::string method_name;  // 方法名
    uint32_t args_size;  // 方法参数长度
    if(rpcHeader.ParseFromString(rpc_header_str))
    {
        // 数据头反序列化成功
        service_name = rpcHeader.service_name();
        method_name = rpcHeader.method_name();
        args_size = rpcHeader.args_size();
    }
    else
    {
        // 数据头反序列化失败
        std::cout << "rpc_header_str: " << rpc_header_str << " parse error!" << std::endl;
        return ;
    }

    // 获取rpc方法参数的字符流数据
    std::string args_str = recv_buf.substr(4 + header_size, args_size);

    // 打印调试信息
    std::cout << "========================================" << std::endl;
    std::cout << "header_size: " << header_size << std::endl;
    std::cout << "rpc_header_str: " << rpc_header_str << std::endl;
    std::cout << "service_name: " << service_name << std::endl;
    std::cout << "method_name: " << method_name << std::endl;
    std::cout << "args_str: " << args_str << std::endl;
    std::cout << "========================================" << std::endl;

    // 获取 service 对象和 method 对象
    auto it = m_serviceMap.find(service_name);
    if(it == m_serviceMap.end())
    {
        std::cout << service_name << " is not exist!" << std::endl;  // 服务对象不存在
        return ;
    }
    auto mit = it->second.m_methodMap.find(method_name);
    if(mit == it->second.m_methodMap.end())
    {
        std::cout << service_name << " : " << method_name << " is not exist!" << std::endl;  // 服务对象存在，方法不存在
        return ;
    }
    google::protobuf::Service *service = it->second.m_service;  // 获取 service 对象  -- new 出来的那个 UserService
    const google::protobuf::MethodDescriptor *method = mit->second;  // 获取 method 对象  -- Login

    // 生成 rpc 方法的调用的 request 请求和响应 response 参数，这两者都是从 Message 继承而来
    google::protobuf::Message *request = service->GetRequestPrototype(method).New();  // 产生某个服务对象的某个方法的请求类型
    if(!request->ParseFromString(args_str))
    {
        std::cout << "request parse error! content: " << args_str << std::endl;
        return ;
    }
    google::protobuf::Message *response = service->GetResponsePrototype(method).New();  // 产生某个服务对象的某个方法的响应类型

    // 给下面的 method 方法，绑定一个 Closure 回调（通过NewCallback产生一个对象）
    google::protobuf::Closure *done = 
        google::protobuf::NewCallback<RpcProvider, 
                                      const muduo::net::TcpConnectionPtr&, 
                                      google::protobuf::Message*>
                                      (this, &RpcProvider::SendRpcResponse, conn, response);
    // 在框架上根据远端 rpc 请求，调用当前 rpc 节点上发布的方法
    service->CallMethod(method, nullptr, request, response, done);  // new UserSerivce().Login(controller, requeset, response, done)
}

// Closure 回调操作，用于序列号 rpc 的响应和网络发送
void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr& conn, google::protobuf::Message* response)
{
    std::string response_str;
    if(response->SerializeToString(&response_str))  // response 序列化
    {
        // 序列化成功后通过网络把rpc方法执行结果发送给rpc的调用方
        conn->send(response_str);
    }
    else
    {
        std::cout << "serialize response_str error." << std::endl;
    }
    conn->shutdown();  // 模拟 http 的短连接服务，由 rpcprovider 主动断开连接，释放资源，给更多的rpc调用方提供资源
}